package com.inspur.springkafka.stream2;

import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.PunctuationType;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.KeyValueStore;

import java.time.Duration;

public class WordCountProcessor implements Processor<String,String> {
    private ProcessorContext context;
    private String storeName;
    private KeyValueStore<String,Integer> stateStore;
    public WordCountProcessor(String storeName) {
        this.storeName = storeName;
    }

    @Override
    public void init(ProcessorContext context) {
        this.context=context;
        stateStore = (KeyValueStore<String, Integer>) context.getStateStore(storeName);
        //定时输出结果
        context.schedule(Duration.ofSeconds(1), PunctuationType.WALL_CLOCK_TIME,(ts)->{
            KeyValueIterator<String, Integer> keyValueIterator = stateStore.all();
            while (keyValueIterator.hasNext()){
                KeyValue<String, Integer> keyValue = keyValueIterator.next();
                context.forward(keyValue.key,keyValue.value);
            }
            keyValueIterator.close();
            context.commit();
        });
    }
    @Override
    public void process(String key, String value) {
        String[] words = value.split("\\W+");
        for (int i = 0; i < words.length; i++) {
            int count=0;
            Integer historyCount = stateStore.get(words[i]);
            if(historyCount!=null){
                count=historyCount;
            }
            count+=1;
            stateStore.put(words[i],count);
        }

    }
    @Override
    public void close() {

    }
}
